package crossjob;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Hashtable;

/**
 * Created by zhaopeng on 2/7/17.
 */
public class CrossJobMapper extends Mapper<LongWritable, Text, Text, Text> {

    @Override
    public void map(LongWritable keys, Text values, Context context) throws IOException, InterruptedException {
        //Text key = new Text(values.toString());
//        if (key.equals(new Text("")))
//            return;
        int[] tags = {562, 563, 564, 565, 566, 567, 568, 569, 570, 571, 12021, 572, 11261,
                11262, 11263, 11264, 11265, 11266, 11267, 11268, 11269, 11270, 11271, 11272, 11273, 11274};
//        Text value = getFileName(((FileSplit) context.getInputSplit()).getPath().toString());
        Hashtable<Integer, Boolean> hashedValue = new Hashtable<Integer, Boolean>();
        for (int i : tags) {
            hashedValue.put(i,true);
        }
        String key = "", value = "";
        String[] strArry = values.toString().split("\t");
        if (1 == strArry.length) {
            key = strArry[0];
            value = "0";
        } else if (strArry.length > 1) {
            value = "";
            key = strArry[0];
            String[] value_tags = strArry[1].split(",");
            for (int i = 0; i < value_tags.length; i++) {
                try {
                    Integer m = Integer.parseInt(value_tags[i]);
                    if (hashedValue.containsKey(m)) {
                        if (i < value_tags.length - 1) {
                            value += value_tags[i] + ",";
                        }else
                        {
                            value += value_tags[i];
                        }
                    }
                } catch (Exception e) {
                    continue;
                }
            }

        }

        context.write(new Text(key), new Text(value));

    }

    /*public Text getFileName(String path) {
        int index = path.lastIndexOf('/') + 1;
        String fileName = path.substring(index);
        return new Text(fileName);
    }*/

}
